package com.sunday.common.reactor.e4_Transforming_Existing_Sequence;

import com.sunday.common.reactor.SubscribeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

/**
 * A.2. 转换现有序列
 */
@Slf4j
public class TransformingExistingSequence {

    /**
     * 在 1 对 1 的基础上（例如，字符串的长度）：（Flux|单map)
     * 只需投射它：（Flux|单cast)
     * 为了具体化每个源值的索引：Flux#index
     */
    @Test
    public void map() {
        Flux<String> flux = Flux.just("foo", "bar", "foobar");
        log.info("map 会将每个源值转换为新值。");
        // map 会将每个源值转换为新值。
        SubscribeUtils.subscribe(flux.map(String::length));
        log.info("cast 会将每个源值转换为新类型。");
        // cast 会将每个源值转换为新类型。
        SubscribeUtils.subscribe(flux.cast(Object.class));
        log.info("index 会将每个源值转换为其索引。");
        // index 会将每个源值转换为其索引。
        SubscribeUtils.subscribe(flux.index());
    }

    /**
     * 以 1 到 n 为基础（例如，字符串到其字符）：（Flux|Mono） + 使用工厂方法flatMap
     */
    @Test
    public void flatMap() {
        Flux<String> flux = Flux.just("foo", "bar", "foobar");
        log.info("flatMap 会将每个源值转换为新序列。");
        // flatMap 会将每个源值转换为新序列。
        SubscribeUtils.subscribe(flux.flatMap(s -> Flux.fromArray(s.split(""))));

        Mono<String> mono = Mono.just("foo");
        log.info("flatMap 会将每个源值转换为新序列。");
        // flatMap 会将每个源值转换为新序列。
        SubscribeUtils.subscribe(mono.flatMap(s -> Mono.just(s + "bar")));
    }

    /**
     * 在 1 到 n 的基础上，每个源元素和/或状态都有编程行为：（Flux|单handle)
     */
    @Test
    public void handle() {
        Flux<String> flux = Flux.just("foo", "bar", "foobar");
        log.info("handle 会将每个源值转换为新序列。");
        // handle 会将每个源值转换为新序列。
        SubscribeUtils.subscribe(flux.handle((s, sink) -> {
            if (s.length() > 3) {
                sink.next(s);
            }
        }));

        Mono<String> mono = Mono.just("foo");
        log.info("handle 会将每个源值转换为新序列。");
        // handle 会将每个源值转换为新序列。
        SubscribeUtils.subscribe(mono.handle((s, sink) -> {
            if (s.length() > 3) {
                sink.next(s);
            }
        }));
    }

    /**
     * 为每个源项运行异步任务（例如，HTTP 请求的 URL）：（Flux|Mono） + 异步发布者返回方法flatMap
     * 忽略一些数据：在 flatMap lambda 中有条件地返回一个 Mono.empty（）
     * 保留原始序列顺序：Flux#flatMapSequential（这会立即触发异步进程，但会对结果重新排序）
     * 异步任务可以从 Mono 源返回多个值：Mono#flatMapMany
     */
    @Test
    public void flatMapAsync() {
        Flux<String> flux = Flux.just("foo", "bar", "foobar");
        log.info("flatMap 会将每个源值转换为新序列。");
        // flatMap 会将每个源值转换为新序列。
        // flatMap 可以并行处理源 Flux 中的元素，不保证输出的顺序与输入一致。
        // 适合于那些可以并行执行的任务，例如独立的数据库查询或远程调用。
        SubscribeUtils.subscribe(flux.flatMap(s -> Mono.just(s + "bar")));

        Mono<String> mono = Mono.just("foo");
        log.info("flatMap 会将每个源值转换为新序列。");
        // flatMap 会将每个源值转换为新序列。
        SubscribeUtils.subscribe(mono.flatMap(s -> Mono.just(s + "bar")));

        log.info("保留原始序列顺序：Flux#flatMapSequential（这会立即触发异步进程，但会对结果重新排序）");
        // 保留原始序列顺序：Flux#flatMapSequential（这会立即触发异步进程，但会对结果重新排序）
        // flatMapSequential 会保持源 Flux 中元素的顺序，并且对每个元素进行序列化的处理。这意味着它会等待前一个转换完成后才会进行下一个转换。
        // 适合那些需要保持顺序的场景，例如处理依赖关系的任务。
        SubscribeUtils.subscribe(flux.flatMapSequential(s -> Mono.just(s + "bar")));

        log.info("异步任务可以从 Mono 源返回多个值：Mono#flatMapMany");
        // 异步任务可以从 Mono 源返回多个值：Mono#flatMapMany
        SubscribeUtils.subscribe(mono.flatMapMany(s -> Flux.just(s + "bar", "abc")));

    }

    /**
     * 我想将预设元素添加到现有序列中：
     * 在开始时：Flux#startWith（T...）
     * 最后：Flux#concatWithValues（T...）
     */
    @Test
    public void startWith() {
        Flux<String> flux = Flux.just("foo", "bar", "foobar");
        log.info("startWith 会在源序列的开头添加元素。");
        // startWith 会在源序列的开头添加元素。
        SubscribeUtils.subscribe(flux.startWith("1", "2"));
        log.info("concatWithValues 会在源序列的结尾添加元素。");
        // concatWithValues 会在源序列的结尾添加元素。
        SubscribeUtils.subscribe(flux.concatWithValues("1", "2"));
    }

    /**
     * 我想聚合一个 Flux：（前缀假设如下）Flux#
     * 放入 List：collectList、collectSortedList
     * 放入 Map：collectMap、collectMultiMap
     * 放入任意容器：collect
     * 进入序列的大小：count
     * 通过在每个元素之间应用一个函数（例如，运行和）： reduce
     * 但是发出每个中间值：scan
     * 从谓词转换为布尔值：
     * 应用于所有值 （AND）：all
     * 应用于至少一个值 （OR）：any。
     * 测试是否存在任何值：hasElements（hasElement 中有一个 Mono 等效项）
     * 测试特定值是否存在：hasElement（T）
     */
    @Test
    public void collect() {

        Flux<String> flux = Flux.just("foo", "bar", "foobar");
        log.info("collectList 会将源序列转换为 List。");
        // collectList 会将源序列转换为 List。
        SubscribeUtils.subscribe(flux.collectList());
        log.info("collectSortedList 会将源序列转换为已排序的 List。");
        // collectSortedList 会将源序列转换为已排序的 List。
        SubscribeUtils.subscribe(flux.collectSortedList());

        log.info("collectMap 会将源序列转换为 Map。");
        // collectMap 会将源序列转换为 Map。
        SubscribeUtils.subscribe(flux.collectMap(s -> s.substring(0, 3)));
        log.info("collectMultiMap 会将源序列转换为 MultiMap。");
        // collectMultiMap 会将源序列转换为 MultiMap。
        SubscribeUtils.subscribe(flux.collectMultimap(s -> s.substring(0, 3)));

        /**
         * 在这里，ArrayList::add 是一个方法引用，它指向 ArrayList 类的 add 方法。这个方法接受一个参数，并且没有返回值。
         * BiConsumer<E, ? super T> 是一个函数式接口，它接受两个参数并且不返回任何结果。在这里，E 表示列表元素的类型，T 是作为输入提供给这个方法的类型。通配符 ? super T 表示"某种类型，该类型是 T 的超类（包括 T 自身）"。
         * 当你将 ArrayList::add 赋给 BiConsumer<E, ? super T> 类型的变量，实际上是将 ArrayList::add 方法作为一个可以接受两个参数的函数，其中第一个参数是 ArrayList 的实例（列表本身），第二个参数是要添加到列表中的元素。因此，ArrayList::add 可以被赋给 BiConsumer<E, ? super T> 类型的变量。
         */
        log.info("collect 会将源序列转换为任意容器。");
        // collect 会将源序列转换为任意容器。
        SubscribeUtils.subscribe(flux.collect(() -> new ArrayList<>(), ArrayList::add));
        log.info("count 会将源序列转换为 long。");
        // count 会将源序列转换为 long。
        SubscribeUtils.subscribe(flux.count());
        log.info("reduce 接受一个初始值和一个二元运算符。它会将初始值应用于第一个源值，然后将结果应用于第二个源值，以此类推。它会发出最终结果。");
        // reduce 接受一个初始值和一个二元运算符。它会将初始值应用于第一个源值，然后将结果应用于第二个源值，以此类推。它会发出最终结果。
        SubscribeUtils.subscribe(flux.reduce((x, y) -> x + "," + y));
        log.info("scan 会将源序列转换为多个值。它会将每个源值应用于给定的函数，并且会将结果应用于下一个值。它会发出每个中间值。");
        // scan 会将源序列转换为多个值。它会将每个源值应用于给定的函数，并且会将结果应用于下一个值。它会发出每个中间值。
        SubscribeUtils.subscribe(flux.scan((x, y) -> x + "," + y));
        log.info("all 会将源序列转换为布尔值。它会将谓词应用于所有源值，并且只有当所有值都返回 true 时才会返回 true。");
        // all 会将源序列转换为布尔值。它会将谓词应用于所有源值，并且只有当所有值都返回 true 时才会返回 true。
        SubscribeUtils.subscribe(flux.all(s -> s.length() > 3));
        log.info("any 会将源序列转换为布尔值。它会将谓词应用于所有源值，并且只有当至少一个值返回 true 时才会返回 true。");
        // any 会将源序列转换为布尔值。它会将谓词应用于所有源值，并且只有当至少一个值返回 true 时才会返回 true。
        SubscribeUtils.subscribe(flux.any(s -> s.length() > 3));
        log.info("hasElements 会将源序列转换为布尔值。它会返回 true，如果源序列至少包含一个元素。");
        // hasElements 会将源序列转换为布尔值。它会返回 true，如果源序列至少包含一个元素。
        SubscribeUtils.subscribe(flux.hasElements());
        log.info("hasElement 会将源序列转换为布尔值。它会返回 true，如果源序列包含给定的元素。");
        // hasElement 会将源序列转换为布尔值。它会返回 true，如果源序列包含给定的元素。
        SubscribeUtils.subscribe(flux.hasElement("foo"));
    }

    /**
     * 我想合并出版商...
     * <p>
     * 按顺序排列：Flux#concat 或 （Flux|单.concatWith(other))
     * 但延迟任何错误，直到发出剩余的发布者：Flux#concatDelayError
     * <p>
     * 但急切地订阅后续发布者：Flux#mergeSequential
     * 按发射顺序（组合项目在出现时发出）：Flux#merge / （Flux|单.mergeWith(other))
     * 具有不同类型（转换合并）：Flux#zip / Flux#zipWith
     * 通过配对值：
     * 从 2 个 Mono 变成一个 Tuple2：Mono#zipWith
     * 当它们全部完成后，来自 n 个 Monos：Mono#zip
     * 通过协调他们的终止：
     * 从 1 Mono 和任何源转换为 Mono<Void>： Mono#and
     * 当它们全部完成时，来自 n 个来源：Mono#when
     * 添加到任意容器类型中：
     * 每次所有边都发出：Flux#zip（最多最小基数）
     * 每次有新值到达任一侧时：Flux#combineLatest
     * 选择第一个发布者...
     * 生成一个值 （）： （Flux|Mono#onNextfirstWithValue)
     * 产生任何信号：（Flux|Mono#firstWithSignal)
     * 由源序列中的元素触发：switchMap（每个源元素都映射到一个 Publisher）
     * 由发布者序列中下一个发布者的开始触发：switchOnNext
     */
    @Test
    public void merge() throws InterruptedException {
        log.info("Flux#concat: 将当前 Observable 与另一个 Observable 进行连接，使得它们的输出按顺序连接起来。如果第一个 Observable 终止了，才会开始发射第二个 Observable 的数据。");
        Flux<String> flux = Flux.just("foo", "bar", "foobar")
                // concatWith: 将当前 Observable 与另一个 Observable 进行连接，使得它们的输出按顺序连接起来。如果第一个 Observable 终止了，才会开始发射第二个 Observable 的数据。
                .concatWith(Flux.generate(() -> 0, (state, sink) -> {
//                    log.info("[state]{}", state);
                    sink.next(state + "===");
                    if (state == 3) {
                        throw new RuntimeException("Some error");
//                        sink.complete();
                    }
                    return state + 1;
                }))
                .concatWith(Flux.just("1", "2"));
        // concatDelayError: 与 concatWith 类似，但是不同之处在于当其中一个 Observable 发生错误时，concatDelayError 会等待所有 Observable 都终止后再将错误通知给观察者；而 concatWith 则会立即将错误通知给观察者。
//                .concatDelayError(Flux.just("error1", "error2"));
        SubscribeUtils.subscribe(flux);

        log.info("Flux#mergeSequential 会将所有源序列的值合并到一个新的序列中。它会按顺序订阅每个源序列，并且会立即订阅下一个序列，而不是等待前一个序列完成。");
        // mergeSequential 会将所有源序列的值合并到一个新的序列中。它会按顺序订阅每个源序列，并且会立即订阅下一个序列，而不是等待前一个序列完成。
        SubscribeUtils.subscribe(Flux.mergeSequential(Flux.just(4, 5, 6), Flux.just(1, 2, 3), Flux.just(7, 8, 9)));

        log.info("Flux#merge 会将所有源序列的值合并到一个新的序列中。它会同时订阅所有源序列，并且会按照源序列发出值的顺序发出值。");
        // merge 会将所有源序列的值合并到一个新的序列中。它会同时订阅所有源序列，并且会按照源序列发出值的顺序发出值。
        SubscribeUtils.subscribe(Flux.merge(Flux.just(4, 5, 6), Flux.just(1, 2, 3)));

        log.info("Flux#zip 会将多个源序列的值合并到一个新的序列中。它会同时订阅所有源序列，并且会将每个源序列的最新值组合成一个元组。");
        // zip 会将多个源序列的值合并到一个新的序列中。它会同时订阅所有源序列，并且会将每个源序列的最新值组合成一个元组。
        SubscribeUtils.subscribe(Flux.zip(Flux.just(4, 5, 6), Flux.just(1, 2, 3)));

        log.info("Flux#zipWith 会将两个源序列的值合并到一个新的序列中。它会同时订阅两个源序列，并且会将每个源序列的最新值组合成一个元组。");
        // zipWith 会将两个源序列的值合并到一个新的序列中。它会同时订阅两个源序列，并且会将每个源序列的最新值组合成一个元组。
        SubscribeUtils.subscribe(Flux.just("foo", "bar", "foobar").zipWith(Flux.just(4, 5, 6, 8, 9).concatWith(Flux.just(10, 20, 21, 22, 23)), (s1, s2) -> s1 + s2));

        log.info("Mono#zipWith 会将两个源序列的值合并到一个新的序列中。它会同时订阅两个源序列，并且会将每个源序列的最新值组合成一个元组。");
        // 从 2 个 Mono 变成一个 Tuple2：Mono#zipWith
        SubscribeUtils.subscribe(Mono.just("foo").zipWith(Mono.just("bar")));

        log.info("Mono#zip 会将多个源序列的值合并到一个新的序列中。它会同时订阅所有源序列，并且会将每个源序列的最新值组合成一个元组。");
        // 当它们全部完成后，来自 n 个 Monos：Mono#zip
        SubscribeUtils.subscribe(Mono.zip(Mono.just("foo"), Mono.just("bar")));

        log.info("Mono#and 会将源序列转换为 Mono<Void>。它会等待源序列完成，然后发出一个空值。");
        // 从 1 Mono 和任何源转换为 Mono<Void>： Mono#and
        SubscribeUtils.subscribe(Mono.just("foo").and(Mono.just("bar")));

        log.info("Mono#when 会将源序列转换为 Mono<Void>。它会等待源序列完成，然后发出一个空值。");
        // 当它们全部完成时，来自 n 个来源：Mono#when
        SubscribeUtils.subscribe(Mono.when(Mono.just("foo"), Mono.just("bar")));

        log.info("Flux#combineLatest 会将多个源序列的值合并到一个新的序列中。它会同时订阅所有源序列，并且会将每个源序列的最新值组合成一个元组。");
        // 每次有新值到达任一侧时：Flux#combineLatest
        SubscribeUtils.subscribe(Flux.combineLatest(Flux.just("foo", "bar", "test"), Flux.just("1", "2", "3"), (s1, s2) -> s1 + s2));

        log.info("Flux#onNext 会将源序列转换为 Mono。它会发出源序列的第一个值，然后立即完成。");
        // 生成一个值 （）：（Flux|单onNext firstWithValue)
        SubscribeUtils.subscribe(Flux.just("foo", "bar", "test").next());

        log.info("Flux.firstWithValue: 这个操作符是用来从多个 Flux 中选择第一个发出元素的 Flux，并忽略那些没有元素的 Flux。如果所有的 Flux 都没有发出元素，那么它就会返回一个空的 Flux。这个操作符类似于找到第一个有值的 Flux。");
        // Flux.firstWithValue: 这个操作符是用来从多个 Flux 中选择第一个发出元素的 Flux，并忽略那些没有元素的 Flux。如果所有的 Flux 都没有发出元素，那么它就会返回一个空的 Flux。这个操作符类似于找到第一个有值的 Flux。
        SubscribeUtils.subscribe(Flux.firstWithValue(Flux.empty(), Flux.just("1", "2", "3")));

        log.info("Flux.firstWithSignal: 与之相反，Flux.firstWithSignal 会从多个 Flux 中选择第一个产生信号（元素或错误）的 Flux。如果所有的 Flux 都没有产生信号，那么它就会返回一个空的 Flux。");
        // Flux.firstWithSignal: 与之相反，Flux.firstWithSignal 会从多个 Flux 中选择第一个产生信号（元素或错误）的 Flux。如果所有的 Flux 都没有产生信号，那么它就会返回一个空的 Flux。
        SubscribeUtils.subscribe(Flux.firstWithSignal(Flux.empty(), Flux.just("1", "2", "3")));


        Flux<Flux<Integer>> fluxOfFlux = Flux.just(
                Flux.just(1, 2, 3).delayElements(Duration.ofSeconds(1)),
                Flux.just(4, 5, 6).delayElements(Duration.ofSeconds(2)),
                Flux.just(7, 8, 9).delayElements(Duration.ofSeconds(3))
        );

        log.info("switchMap: 在 Reactor 中，switchMap 用于将源 Flux 的每个元素转换为一个新的 Flux，并且只发出最近产生的新 Flux 所产生的元素。这意味着如果在处理当前元素时，新的元素到达，它会取消订阅先前的转换后的 Flux，并开始订阅最近的转换后的 Flux。");
        // switchMap: 在 Reactor 中，switchMap 用于将源 Flux 的每个元素转换为一个新的 Flux，并且只发出最近产生的新 Flux 所产生的元素。这意味着如果在处理当前元素时，新的元素到达，它会取消订阅先前的转换后的 Flux，并开始订阅最近的转换后的 Flux。
        SubscribeUtils.subscribe(fluxOfFlux.switchMap(s -> Flux.just(s.blockFirst() + "bar")));

        log.info("switchOnNext: 相比之下，switchOnNext 是用于将包含其他 Flux 的 Flux（也称为 '高阶 Flux'）转换为单一的 Flux。每当外部的 Flux 发出一个新的内部 Flux 时，switchOnNext 会取消订阅先前的内部 Flux，并开始订阅最新的内部 Flux。");
        // switchOnNext: 相比之下，switchOnNext 是用于将包含其他 Flux 的 Flux（也称为 "高阶 Flux"）转换为单一的 Flux。每当外部的 Flux 发出一个新的内部 Flux 时，switchOnNext 会取消订阅先前的内部 Flux，并开始订阅最新的内部 Flux。

        Flux<Integer> switched = Flux.switchOnNext(fluxOfFlux);

        SubscribeUtils.subscribe(switched);

        TimeUnit.SECONDS.sleep(10);

    }

    /**
     * 我想重复现有的序列：（Flux|Mono#repeat)
     * 但在时间间隔内：Flux.interval(duration).flatMap(tick → myExistingPublisher)
     */
    @Test
    public void repeat() {
        Flux<String> flux = Flux.just("foo", "bar", "foobar");
        log.info("repeat 会将源序列重复 n 次。");
        // repeat 会将源序列重复 n 次。
        SubscribeUtils.subscribe(flux.repeat(2));

        log.info("repeatWhen 会将源序列重复 n 次。");
        // repeatWhen 会将源序列重复 n 次。
        SubscribeUtils.subscribe(flux.repeatWhen(completed -> completed.take(2)));
    }

    /**
     * 我有一个空序列，但是......
     * 我想要一个值：（Flux|单defaultIfEmpty)
     * 我想要另一个序列：（Flux|单switchIfEmpty)
     */
    @Test
    public void defaultIfEmpty() {
        Flux<String> flux = Flux.empty();
        log.info("defaultIfEmpty 会将源序列转换为一个新序列。如果源序列为空，它会发出一个默认值。");
        // defaultIfEmpty 会将源序列转换为一个新序列。如果源序列为空，它会发出一个默认值。
        SubscribeUtils.subscribe(flux.defaultIfEmpty("foo"));

        log.info("switchIfEmpty 会将源序列转换为一个新序列。如果源序列为空，它会发出一个新的序列。");
        // switchIfEmpty 会将源序列转换为一个新序列。如果源序列为空，它会发出一个新的序列。
        SubscribeUtils.subscribe(flux.switchIfEmpty(Mono.just("foo")));
    }

    /**
     * 我有一个序列，但我对值不感兴趣：（Flux.ignoreElements（）|Mono.ignoreElement（）ignoreElements)
     * 我希望完成表示为 Mono<Void>：（Flux|Mono#then)
     * 我想等待最后完成另一个任务：（Flux|Mono#thenEmpty)
     * 我想在最后切换到另一个 Mono：Mono#then（mono）
     * 我想在最后发出一个值：Mono#thenReturn（T）
     * 我想在最后切换到 Flux：（Flux|Mono#thenMany)
     */
    @Test
    public void ignoreElements() {

        Flux<String> flux = Flux.just("foo", "bar", "foobar").doOnNext(s -> log.info("doOnNext:{}", s));
        log.info("ignoreElements 会将源序列转换为 Mono。它会忽略源序列的所有值，只发出完成信号。");
        // ignoreElements 会将源序列转换为 Mono。它会忽略源序列的所有值，只发出完成信号。
        SubscribeUtils.subscribe(flux.ignoreElements());

        log.info("flux.then: then 操作符允许你在一个 Flux 完成后执行另一个操作。当原始 Flux 完成时，then 允许你执行一个回调或返回另一个 Mono 或 Flux，以便进行进一步的处理");
        // 我希望完成表示为 Mono<Void>：（Flux|单then) flux.then: then 操作符允许你在一个 Flux 完成后执行另一个操作。当原始 Flux 完成时，then 允许你执行一个回调或返回另一个 Mono 或 Flux，以便进行进一步的处理
        SubscribeUtils.subscribe(flux.then(Mono.just("test")));

        log.info("flux.thenEmpty: 与之相反，thenEmpty 也是在原始 Flux 完成后执行操作，但它返回的是一个空的 Mono。它通常用于在 Flux 完成后执行一些操作，而不需要发出任何值");
        // 我想在最后切换到另一个 Mono：Mono#then（mono） flux.thenEmpty: 与之相反，thenEmpty 也是在原始 Flux 完成后执行操作，但它返回的是一个空的 Mono。它通常用于在 Flux 完成后执行一些操作，而不需要发出任何值
        SubscribeUtils.subscribe(flux.thenEmpty(Mono.empty()));

        log.info("thenMany 允许你在一个 Flux 完成时执行另一个操作并返回另一个 Flux。这个操作符通常用于在原始 Flux 完成后继续执行另一个 Flux，并将其结果作为整体的结果。");
        // 我想在最后切换到 Flux：（Flux|单thenMany) thenMany 会将源序列转换为 Mono。它会忽略源序列的所有值，只发出完成信号。 thenMany 允许你在一个 Flux 完成时执行另一个操作并返回另一个 Flux。这个操作符通常用于在原始 Flux 完成后继续执行另一个 Flux，并将其结果作为整体的结果。
        SubscribeUtils.subscribe(flux.thenMany(Mono.just("1")));

        log.info("thenMany 允许你在一个 Flux 完成时执行另一个操作并返回另一个 Flux。这个操作符通常用于在原始 Flux 完成后继续执行另一个 Flux，并将其结果作为整体的结果。");
        // 我想在最后切换到 Flux：（Flux|单thenMany) thenMany 会将源序列转换为 Mono。它会忽略源序列的所有值，只发出完成信号。 thenMany 允许你在一个 Flux 完成时执行另一个操作并返回另一个 Flux。这个操作符通常用于在原始 Flux 完成后继续执行另一个 Flux，并将其结果作为整体的结果。
        SubscribeUtils.subscribe(flux.thenMany(Flux.just("1", "2", "3")));

        log.info("Mono#thenReturn 是一个操作符，它用于创建一个只发出指定值的 Mono。一旦订阅了这个 Mono，它将立即发出指定的值，然后完成。这个操作符主要用于将一个常量值转换为 Mono。");
        // Mono#thenReturn 是一个操作符，它用于创建一个只发出指定值的 Mono。一旦订阅了这个 Mono，它将立即发出指定的值，然后完成。这个操作符主要用于将一个常量值转换为 Mono。
        SubscribeUtils.subscribe(Mono.just("first").thenReturn("last"));

    }

    /**
     * 我有一个单声道，我想推迟完成......
     * 直到从此值派生的另一个发布者完成：Mono#delayUntil（Function）
     */
    @Test
    public void delayUntil() throws InterruptedException {

        Mono<String> mono = Mono.just("foo").delayElement(Duration.ofSeconds(3));
        Mono<String> delay = Mono.just("bar").delayElement(Duration.ofSeconds(3));
        log.info("delayUntil 会将源序列转换为 Mono。它会推迟完成，直到另一个 Mono 完成。");
        // delayUntil 会将源序列转换为 Mono。它会推迟完成，直到另一个 Mono 完成。
        SubscribeUtils.subscribe(mono.delayUntil(s -> delay));

        TimeUnit.SECONDS.sleep(10);
    }

    /**
     * 我想以递归方式将元素扩展为序列图并发出组合......
     * 首先扩展图的宽度：（Flux|Mono#expand(Function))
     * 首先扩展图形深度：（Flux|Mono#expandDeep(Function))
     */
    @Test
    public void expand() {
        Flux<Integer> flux = Flux.just(1, 10, 20);
        log.info("expand 方法用于递归展开 Flux 序列中的元素，类似于展开一颗树的节点。它接收一个回调函数，该函数返回一个 Flux，用来生成新的元素，这些元素会被添加到结果序列中。");
        // expand 会将源序列转换为新序列。它会将每个源值转换为一个新序列，并将这些序列的值合并到一个新序列中。
        // 方法用于递归展开 Flux 序列中的元素，类似于展开一颗树的节点。它接收一个回调函数，该函数返回一个 Flux，用来生成新的元素，这些元素会被添加到结果序列中。
        SubscribeUtils.subscribe(flux.expand(value -> {
            if (value < 10) {
                return Flux.just(value + 1);
            } else {
                return Flux.empty();
            }
        }));

        log.info("expandDeep 方法也递归地展开 Flux 序列中的元素，与 expand 不同之处在于它通过额外的参数控制展开的深度，以避免无限递归。");
        // expandDeep 会将源序列转换为新序列。它会将每个源值转换为一个新序列，并将这些序列的值合并到一个新序列中。
        // 方法也递归地展开 Flux 序列中的元素，与 expand 不同之处在于它通过额外的参数控制展开的深度，以避免无限递归。
        SubscribeUtils.subscribe(flux.expandDeep(value -> {
            if (value < 10) {
                return Flux.just(value + 1);
            } else {
                return Flux.empty();
            }
        }));
    }

}
